package com.mjf.day2

import org.apache.flink.streaming.api.scala._

/**
 * keyBy
 *  基于key的hash code重分区；
 *  同一个key只能在一个分区内处理，一个分区内可以有不同的key的数据
 *  keyBy之后的KeyedStream上的所有操作，针对的作用域都只是当前的key
 *
 * 滚动聚合操作
 *  DataStream没有聚合操作，目前所有的聚合操作都是针对KeyedStream
 */
object KeyByStreamExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[SensorReading] = env.addSource(new SensorSource).filter(_.id.equals("sensor_1"))

    // 第二个泛型是key的类型
    val keyedStream: KeyedStream[SensorReading, String] = stream.keyBy(_.id)
//    val keyedStream: KeyedStream[SensorReading, String] = stream.keyBy(0)
//    val keyedStream: KeyedStream[SensorReading, String] = stream.keyBy("id")

    // 使用第三个字段来做滚动聚合，求每个传感器流上的最小温度值
    // 内部会保存一个最小值的状态变量，用来保存到来的温度的最小值
//    keyedStream.min(2).print()

    // reduce也会保存一个状态变量
    keyedStream
      .reduce((s1, s2) => SensorReading(s1.id, s1.timestamp.max(s2.timestamp), s1.temperature.min(s2.temperature)))
      .print()

    env.execute("KeyByStreamExample")

  }
}
